package com.simple.trade.service.vm.impl;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.base.*;
import com.google.common.collect.Iterators;
import com.google.common.collect.Lists;
import com.google.common.primitives.Longs;
import com.simple.common.entity.dto.pay.*;
import com.simple.common.entity.po.sys.*;
import com.simple.ds.service.*;
import com.simple.trade.holder.PayContext;
import com.simple.trade.service.vm.CallbackHandler;
import com.simple.trade.service.vm.DistributeService;
import com.simple.common.entity.ResData;
import com.simple.common.entity.po.NotifyChannel;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.client.ServiceInstance;
import org.springframework.cloud.client.loadbalancer.LoadBalancerClient;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;

import javax.servlet.http.HttpServletRequest;
import java.io.IOException;
import java.util.Objects;
import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import static com.simple.common.entity.constant.Payment.*;


/**
 * @author yasud
 * @date 2020年10月12日 09:22
 * @description
 */
@Slf4j
@Service
public class CallbackHandlerImpl implements CallbackHandler {

    @Autowired
    ChannelService channelService;

    @Autowired
    TradeApi tradeApi;

    @Autowired
    RefundTradeInventoryApi refundTradeInventoryApi;

    @Autowired
    RefundTradeApi refundTradeApi;

    @Autowired
    NotifyChannelService notifyChannelApi;

    @Autowired
    ObjectMapper objectMapper;

    @Autowired
    LoadBalancerClient loadBalancerClient;

    @Autowired
    RestTemplate restTemplate;

    @Autowired
    DistributeService distributeService;

    /**
     * 处理固定地址回调
     *
     * @param request
     * @param channelFlag
     * @param isRefund
     * @param param
     * @return
     * @throws IOException
     */
    @Override
    public String handleFix(HttpServletRequest request, String channelFlag, Boolean isRefund, Map<String, Object> param) throws IOException {
        ChannelComp channel = channelService.findChannelCompByFlag(channelFlag);
        Verify.verifyNotNull(channel, "渠道标记%s无效请检查", channelFlag);
        if (param.containsKey("channelMerchantId")) {
            channel = channelService.flexibility(Longs.tryParse(param.get("channelMerchantId") + ""));
        }
        NotifyChannel notifyChannel = notifyChannelApi.save_(generateChannelNotify(request, NotifyChannel.Type.FIX, isRefund, null, null, null, channel, null,true));
//        eventBus.pushTo(AmqpConstant.Amqp.CALL, objectMapper.writeValueAsString(notifyChannel));
        distributeService.distribute(notifyChannel);
        return processResult(channel, notifyChannel);
    }

    /**
     * 处理webhook
     *
     * @param request
     * @param channelFlag
     * @return
     * @throws IOException
     */
    @Override
    public String handleWebhook(HttpServletRequest request, String channelFlag, Map<String, Object> pathParam) throws IOException {
        ChannelComp channel = channelService.findChannelCompByFlag(channelFlag);
        Verify.verifyNotNull(channel, "渠道标记%s无效请检查", channelFlag);
        if (pathParam.containsKey("channelMerchantId")) {
            channel = channelService.flexibility(Longs.tryParse(pathParam.get("channelMerchantId") + ""));
        }
        NotifyChannel t = generateChannelNotify(request, NotifyChannel.Type.FIX, false, null, null, null, channel, null,true);
        CallbackFixWebhookDto callbackFixWebhookDto = generateWebhookDto(t, channel);
        ResData<CallbackFixWebhookRspDto> rsp = forwardToChannel(callbackFixWebhookDto, channel, CALLBACK_FIX_WEBHOOK, CallbackFixWebhookRspDto.class);
        if (rsp.isOk()) {
            CallbackFixWebhookRspDto data = rsp.getData();
            if (data.getType() == CallbackFixWebhookRspDto.TRADE_TYPE.TRADE) {
//                NotifyChannel t = generateChannelNotify(request, NotifyChannel.Type.FIX, false, null, data.getTradeNo(), null, channel);
                if (Objects.isNull(data.getTradeNo())) {
                    Preconditions.checkState(StringUtils.isNotBlank(data.getChannelTradeNo()), "在未获得平台交易id的情况下，必须返回渠道交易id");
                    List<Trade> trades = tradeApi.findTradeByChannelOrderNo(data.getChannelTradeNo());
                    if (CollectionUtils.isNotEmpty(trades) && trades.size() > 1) {
                        log.info("注意渠道订单{}查询出大于一条平台, 系统默认选择最新一条", data.getChannelTradeNo());
                    }
                    Trade trade = trades.stream().sorted(Comparator.comparing(BaseEntity2::getCreateTime).reversed()).findFirst().orElse(null);
                    Preconditions.checkNotNull(trade, "找不到渠道交易id{}的交易", data.getChannelTradeNo());
                    t.setIsRefund(false);
                    t.setTradeId(trade.getId());
                    NotifyChannel notifyChannel = notifyChannelApi.save_(t);
//                    eventBus.pushTo(AmqpConstant.Amqp.CALL, objectMapper.writeValueAsString(notifyChannel));
                    distributeService.distribute(notifyChannel);
                    return processResult(channel, notifyChannel);
                }
                else {
                    t.setIsRefund(false);
                    t.setTradeId(data.getTradeNo());
                    NotifyChannel notifyChannel = notifyChannelApi.save_(t);
//                    eventBus.pushTo(AmqpConstant.Amqp.CALL, objectMapper.writeValueAsString(notifyChannel));
                    distributeService.distribute(notifyChannel);
                    return processResult(channel, notifyChannel);
                }
            } else if (data.getType() == CallbackFixWebhookRspDto.TRADE_TYPE.REFUND) {
//                NotifyChannel t = generateChannelNotify(request, NotifyChannel.Type.FIX, true, data.getRefundTradeInventoryNo(), null, null, channel);
                if (Objects.isNull(data.getRefundTradeInventoryNo())) {
                    Preconditions.checkState(StringUtils.isNotBlank(data.getChannelRefundTradeNo()), "在未获得平台退款id的情况下，必须返回渠道退款id");
                    List<RefundTradeInventory> refundTrades = refundTradeInventoryApi.findByChannelRefundNo(data.getChannelRefundTradeNo());
                    if (CollectionUtils.isNotEmpty(refundTrades) && refundTrades.size() > 1) {
                        log.info("注意渠道退款订单{}查询出大于一条, 系统默认选择最新一条", data.getChannelTradeNo());
                    }
                    RefundTradeInventory inventory = refundTrades.stream().sorted(Comparator.comparing(BaseEntity2::getCreateTime).reversed()).findFirst().orElse(null);
                    Preconditions.checkNotNull(inventory, "找不到渠道退款id{}的退款", data.getChannelRefundTradeNo());
                    t.setIsRefund(true);
                    t.setRefundTradeInventoryId(inventory.getId());
                    NotifyChannel notifyChannel = notifyChannelApi.save_(t);
//                    eventBus.pushTo(AmqpConstant.Amqp.CALL, objectMapper.writeValueAsString(notifyChannel));
                    distributeService.distribute(notifyChannel);
                    return processResult(channel, notifyChannel);
                }
                else {
                    t.setIsRefund(true);
                    t.setRefundTradeInventoryId(data.getRefundTradeInventoryNo());
                    NotifyChannel notifyChannel = notifyChannelApi.save_(t);
//                    eventBus.pushTo(AmqpConstant.Amqp.CALL, objectMapper.writeValueAsString(notifyChannel));
                    distributeService.distribute(notifyChannel);
                    return processResult(channel, notifyChannel);
                }
            } else {
                Preconditions.checkState(false, "解析失败 无效 订单类型");
                return "";
            }
        }
        Preconditions.checkState(false, "解析失败" + rsp.getMsg());
        return "";
    }

    @Override
    public String handleDynamic(HttpServletRequest request, Long tradeNo, Boolean flag, Long refundTradeInventory, Map<String, Object> param) throws IOException {
        if (Objects.nonNull(tradeNo)) {
            Trade trade = tradeApi.findById_(tradeNo);
            Verify.verifyNotNull(trade, "交易%s无效,请检查", tradeNo);
            PayContext.current().iam(trade);
            ChannelComp channel = PayContext.current().channel();
            NotifyChannel notifyChannel = notifyChannelApi.save_(generateChannelNotify(request, NotifyChannel.Type.DYNAMIC, false, null, tradeNo, PayContext.current().exp().getAppId(), channel, param,flag));
//            eventBus.pushTo(AmqpConstant.Amqp.CALL, objectMapper.writeValueAsString(notifyChannel));
            distributeService.distribute(notifyChannel);
            //一些渠道会出现特殊情况
            return processResult(channel, notifyChannel);
        } else if (Objects.nonNull(refundTradeInventory)) {
            RefundTradeInventory tradeInventory = refundTradeInventoryApi.findById(refundTradeInventory);
            Verify.verifyNotNull(tradeInventory, "退款%s无效,请检查", refundTradeInventory);
            RefundTrade refundTrade = refundTradeApi.findByTradeId(tradeInventory.getRefundTradeId());
            Trade trade = tradeApi.findById_(refundTrade.getTradeId());
            PayContext.current().iam(trade);
            ChannelComp channel = PayContext.current().channel();
            NotifyChannel notifyChannel = notifyChannelApi.save_(generateChannelNotify(request, NotifyChannel.Type.DYNAMIC, true, refundTradeInventory, trade.getId(), PayContext.current().exp().getAppId(), channel, null,flag));
//            eventBus.pushTo(AmqpConstant.Amqp.CALL, objectMapper.writeValueAsString(notifyChannel));
            distributeService.distribute(notifyChannel);
            //一些渠道会出现特殊情况
            return processResult(channel, notifyChannel);
        } else {
            Preconditions.checkState(false, "参数无效");
            return null;
        }
    }

    @SneakyThrows
    private CallbackFixWebhookDto generateWebhookDto(NotifyChannel notifyChannel, ChannelComp channel) {
        CallbackFixWebhookDto callbackFixWebhookDto = CallbackFixWebhookDto.generateCallbackDto(notifyChannel.getBody(), notifyChannel.readMapFromParam(), notifyChannel.readMapFromHeader(), new Date(), channel, null, null);
        return callbackFixWebhookDto;
    }

    /**
     * 处理字段传递地址 回调
     *
     * @param request
     * @param tradeNo
     * @return
     * @throws IOException
     */
    @Override
    public String handleDynamic(HttpServletRequest request, Long tradeNo, Long refundTradeInventory, Map<String, Object> param) throws IOException {
        if (Objects.nonNull(tradeNo)) {
            Trade trade = tradeApi.findById_(tradeNo);
            Verify.verifyNotNull(trade, "交易%s无效,请检查", tradeNo);
            PayContext.current().iam(trade);
            ChannelComp channel = PayContext.current().channel();
            NotifyChannel notifyChannel = notifyChannelApi.save_(generateChannelNotify(request, NotifyChannel.Type.DYNAMIC, false, null, tradeNo, PayContext.current().exp().getAppId(), channel, param,true));
//            eventBus.pushTo(AmqpConstant.Amqp.CALL, objectMapper.writeValueAsString(notifyChannel));
            distributeService.distribute(notifyChannel);
            //一些渠道会出现特殊情况
            return processResult(channel, notifyChannel);
        } else if (Objects.nonNull(refundTradeInventory)) {
            RefundTradeInventory tradeInventory = refundTradeInventoryApi.findById(refundTradeInventory);
            Verify.verifyNotNull(tradeInventory, "退款%s无效,请检查", refundTradeInventory);
            RefundTrade refundTrade = refundTradeApi.findByTradeId(tradeInventory.getRefundTradeId());
            Trade trade = tradeApi.findById_(refundTrade.getTradeId());
            PayContext.current().iam(trade);
            ChannelComp channel = PayContext.current().channel();
            NotifyChannel notifyChannel = notifyChannelApi.save_(generateChannelNotify(request, NotifyChannel.Type.DYNAMIC, true, refundTradeInventory, trade.getId(), PayContext.current().exp().getAppId(), channel, null,false));
//            eventBus.pushTo(AmqpConstant.Amqp.CALL, objectMapper.writeValueAsString(notifyChannel));
            distributeService.distribute(notifyChannel);
            //一些渠道会出现特殊情况
            return processResult(channel, notifyChannel);
        } else {
            Preconditions.checkState(false, "参数无效");
            return null;
        }
    }


    /**
     * 处理最终结果
     *
     * @param channel
     * @param notifyChannel
     * @return
     */
    @SneakyThrows
    private String processResult(ChannelComp channel,
                                 NotifyChannel notifyChannel) {
        if (StringUtils.isNotBlank(channel.getChannelNotifyOkRes())) {
            log.info("渠道{}已定义channelNotifyResOk直接返回", channel.getServiceProvider());
            return channel.getChannelNotifyOkRes();
        }
        CallbackResultTradeNoDto tradeNoDto = CallbackResultTradeNoDto.generateCallbackDto(notifyChannel.getBody(), notifyChannel.readMapFromParam(), notifyChannel.readMapFromHeader(), notifyChannel.getCreateTime(), channel);
        ResData<CallbackResultTradeNoRspDto> tradeNoRspDtoResData = forwardToChannel(tradeNoDto, channel, CALLBACK_RES_ORDER_CODE, CallbackResultTradeNoRspDto.class);
        if (tradeNoRspDtoResData.isOk()) {
            log.info("获取交易id成功{}", objectMapper.writeValueAsString(tradeNoRspDtoResData));
            CallbackResultTradeNoRspDto data = tradeNoRspDtoResData.getData();
            Long tradeNo = data.getTradeNo();
            Verify.verifyNotNull(tradeNo, "返回结果失败渠道服务%s在没有定义channelNotifyOkRes的情况下必须返回tradeNo", channel.getServiceProvider());
            Trade trade = tradeApi.findById_(tradeNo);
            Verify.verifyNotNull(trade, "交易%s无效", tradeNo);
            PayContext.current().iam(trade);
            channel = PayContext.current().channel();
            Product product = PayContext.current().product();

            CallbackResultDto callbackResultDto = CallbackResultDto.generateCallbackDto(notifyChannel.getBody(), notifyChannel.readMapFromParam(), notifyChannel.readMapFromHeader(), trade, notifyChannel.getCreateTime(), channel, PayContext.current().getContext(), product);
            ResData<CallbackResultRspDto> resultRspDtoResData = forwardToChannel(callbackResultDto, channel, CALLBACK_RES_CODE, CallbackResultRspDto.class);

            if (resultRspDtoResData.isOk()) {
                log.info("获取结果成功{}", objectMapper.writeValueAsString(resultRspDtoResData));
                CallbackResultRspDto rData = resultRspDtoResData.getData();
                String result = rData.getResult();
                return result;
            } else {
                Verify.verify(false, "获取结果失败 %s", tradeNoRspDtoResData.getMsg());
                return null;
            }
        } else {
            Verify.verify(false, "获取交易id失败 %s", tradeNoRspDtoResData.getMsg());
            return null;
        }
    }

    /**
     * 这里用rest方式最简单，
     * 内部接口相互通信全部使用post
     *
     * @param directPayDto
     * @param channel
     */
    @SneakyThrows
    public <R> ResData<R> forwardToChannel(BaseDto directPayDto, ChannelComp channel, String type, Class<R> responseClass) {
        log.info("请求渠道{}开始参数为{}", channel.getCode(), objectMapper.writeValueAsString(directPayDto));
        ServiceInstance instance = loadBalancerClient.choose(channel.getServiceProvider());
        Preconditions.checkNotNull(instance, "The service is starting. Please try again later");
        ResData resData = Suppliers.memoize(() -> {
            try {
                ResponseEntity<String> responseEntity = restTemplate.postForEntity(getServiceUrl(instance, type), directPayDto, String.class);
                ResData rResData = objectMapper.readValue(responseEntity.getBody(), ResData.class);
                return rResData;
            } catch (Exception ex) {
                ex.printStackTrace();
                return ResData.error(Throwables.getRootCause(ex).getMessage());
            }
        }).get();
        R r = objectMapper.readValue(objectMapper.writeValueAsString(resData.getData()), responseClass);
        resData.setData(r);
        log.info("请求渠道{}结束返回{}", channel.getCode(), objectMapper.writeValueAsString(resData));
        return resData;
    }

    private String getServiceUrl(ServiceInstance instance, String type) {
        return "http://" + instance.getServiceId() + "/" + type;
    }

    /**
     * 重新发送
     *
     * @param notifyChannelId
     * @throws JsonProcessingException
     */
    @Override
    public void reSendNotify(Long notifyChannelId) throws JsonProcessingException {
        NotifyChannel notifyChannel = notifyChannelApi.findById_(notifyChannelId);
//        eventBus.pushTo(AmqpConstant.Amqp.CALL, objectMapper.writeValueAsString(notifyChannel));
        distributeService.distribute(notifyChannel);
    }

    /**
     * 创建
     *
     * @param request
     * @param isRefund
     * @param appId
     * @param channel
     * @return
     * @throws IOException
     */
    public NotifyChannel generateChannelNotify(HttpServletRequest request,
                                               NotifyChannel.Type type,
                                               Boolean isRefund,
                                               Long refundTradeInventory,
                                               Long tradeId,
                                               String appId,
                                               ChannelComp channel,
                                               Map<String, Object> param,
                                               Boolean isSendMerchant) throws IOException {
        NotifyChannel notifyChannel = new NotifyChannel();
        notifyChannel.setIsRefund(isRefund);
        notifyChannel.setAppId(appId);
        notifyChannel.setBody(getBody(request));
        notifyChannel.setChannelId(channel.getId());
        notifyChannel.setHeader(getHeaderString(request));
        notifyChannel.setParam(getParamString(request, param));
        notifyChannel.setIsProcessed(false);
        notifyChannel.setIsSendMerchant(isSendMerchant);
        notifyChannel.setType(type);
        notifyChannel.setTradeId(tradeId);
        notifyChannel.setRefundTradeInventoryId(refundTradeInventory);
        return notifyChannel;
    }

    /**
     * 得到请求体
     *
     * @param request
     * @return
     * @throws IOException
     */
    private String getBody(HttpServletRequest request) throws IOException {
        String s = IOUtils.toString(request.getInputStream(), "utf-8");
        return s;
    }

    /**
     * 获取参数
     *
     * @param request
     * @return
     */
    private String getParamString(HttpServletRequest request) throws JsonProcessingException {
        Map<String, Object> collect = getParam(request);
        String s = objectMapper.writeValueAsString(collect);
        return s;
    }

    /**
     * 获取参数
     *
     * @param request
     * @return
     */
    private String getParamString(HttpServletRequest request, Map<String, Object> param) throws JsonProcessingException {
        Map<String, Object> collect = getParam(request);
        if (MapUtils.isNotEmpty(param)) {
            collect.putAll(param);
        }
        String s = objectMapper.writeValueAsString(collect);
        return s;
    }


    /**
     * 获取参数map
     *
     * @param request
     * @return
     */
    private Map<String, Object> getParam(HttpServletRequest request) {
        Stream<Map.Entry<String, String[]>> stream = request.getParameterMap().entrySet().stream();
        Map collect = stream.filter(stringEntry -> {
            String[] value = stringEntry.getValue();
            return Objects.nonNull(value) && value.length >= 1;
        }).map(stringEntry -> {
            String key = stringEntry.getKey();
            String[] value = stringEntry.getValue();
            if (value.length > 1) {
                return new <String, String>AbstractMap.SimpleEntry(key, value);
            } else {
                return new <String, String>AbstractMap.SimpleEntry(key, value[0]);
            }
        }).collect(Collectors.toMap(entry -> entry.getKey(), entry -> entry.getValue()));
        return collect;
    }

    /**
     * 得到头
     *
     * @param request
     * @return
     * @throws JsonProcessingException
     */
    private String getHeaderString(HttpServletRequest request) throws JsonProcessingException {
        Map<String, Object> collect = getHeaders(request);
        String s = objectMapper.writeValueAsString(collect);
        return s;
    }

    private Map<String, Object> getHeaders(HttpServletRequest request) {
        Enumeration<String> headerNames = request.getHeaderNames();
        Map collect = Lists.newArrayList(Iterators.forEnumeration(headerNames)).stream().filter(s -> {
            return StringUtils.isNotBlank(request.getHeader(s));
        }).map(s -> {
            return new AbstractMap.SimpleEntry(s, request.getHeader(s));
        }).collect(Collectors.toMap(simpleEntry -> simpleEntry.getKey(), simpleEntry -> simpleEntry.getValue()));
        return collect;
    }
}
